package top.cyuw.simplerpc.registry.zookeeper;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import top.cyuw.simplerpc.config.RpcConfig;
import top.cyuw.simplerpc.context.RpcContext;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author chen
 * @date 2023/3/17 09:17
 */
public class ZookeeperClientUtils {

    public static final String ROOT_PATH = "/simple-rpc";

    private static volatile CuratorFramework zkClient;

    private ZookeeperClientUtils() {

    }

    public static CuratorFramework createZkClient() {
        if (zkClient == null) {
            synchronized (ZookeeperClientUtils.class) {
                if (zkClient == null) {
                    RpcConfig rpcConfig = RpcContext.getInstance().getRpcConfig();
                    zkClient = createZkClient(rpcConfig.getZookeeperIp() + ":" + rpcConfig.getZookeeperPort());
                }
            }
        }
        return zkClient;
    }

    public static CuratorFramework createZkClient(String address) {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework zkClient = CuratorFrameworkFactory.builder()
                .connectString(address)
                .retryPolicy(retryPolicy)
                .build();
        zkClient.start();
        try {
            if (!zkClient.blockUntilConnected(30, TimeUnit.SECONDS)) {
                throw new RuntimeException("time out waiting to connect to zk.");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return zkClient;
    }

    public static void createEphemeralNode(CuratorFramework zkClient, String path, String value) throws Exception {
        zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL)
                .forPath(path, value.getBytes(StandardCharsets.UTF_8));
    }

    public static List<String> getChildrenNodes(CuratorFramework zkClient, String path) throws Exception {
        return zkClient.getChildren().forPath(path);
    }

    public static String getNodeData(CuratorFramework zkClient, String path) throws Exception {
        return new String(zkClient.getData().forPath(path), StandardCharsets.UTF_8);
    }

    public static void registerWatcher(CuratorFramework zkClient, String path, PathChildrenCacheListener listener) throws Exception {
        PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, path, true);
        pathChildrenCache.getListenable().addListener(listener);
        pathChildrenCache.start();
    }

    public static void deleteNode(CuratorFramework zkClient, String path) throws Exception {
        zkClient.delete().forPath(path);
    }

}
